package com.lab.hbase.labAPI;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.junit.Before;

/**
 * mapreduce操作hbase
 * 
 */
public class HBaseMr {
	/**
	 * 配置ss
	 */
	static Configuration config = null;
	private Connection connection = null;
	private static Table table = null;
	// private Admin admin = null;
	@Before
	public void init() throws Exception {
		System.out.println("##连接初始化");
		// 创建hbase配置信息，默认读取hbase-site.xml文件内容进行配置
		config = HBaseConfiguration.create();
		System.out.println("配置文件：" + config);
		// 修改该配置信息
		// config.set("hbase.zookeeper.quorum", "slave1,slave2,slave3");// zookeeper地址
		// config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口

		// 获取hbase连接
		connection = ConnectionFactory.createConnection(config);
		System.out.println("连接是否关闭： " + connection.isClosed());

		// 获取待处理的table对象
		table = connection.getTable(TableName.valueOf(tableName2));
		System.out.println("获取表：" + table);

	}
	
	/**
	 * 表信息
	 */
	public static final String tableName = "word";// 表名1
	public static final String colf = "content";// 列族
	public static final String col = "info";// 列
	public static final String tableName2 = "stat";// 表名2

	public static void main(String[] args) throws Exception {
		 HBaseMr hbmr = new HBaseMr();
		 hbmr.init();
		 //config.set("df.default.name", "hdfs://master:9000/");//设置hdfs的默认路径
		// config.set("hadoop.job.ugi", "hadoop,hadoop");//用户名，组
		// config.set("mapred.job.tracker", "master:9001");//设置jobtracker在哪

		// 创建job
		Job job = new Job(config, "HBaseMr");// job
		job.setJarByClass(HBaseMr.class);// 主类
		// 创建scan
		Scan scan = new Scan();
		// 可以指定查询某一列
		scan.addColumn(Bytes.toBytes(colf), Bytes.toBytes(col));
		// 创建查询hbase的mapper，设置表名、scan、mapper类、mapper的输出key、mapper的输出value
		TableMapReduceUtil.initTableMapperJob(tableName, scan, MyMapper.class, Text.class, IntWritable.class, job);
		// 创建写入hbase的reducer，指定表名、reducer类、job
		TableMapReduceUtil.initTableReducerJob(tableName2, MyReducer.class, job);
		
		System.out.println("mapreduce任务开始……");
		boolean complet =job.waitForCompletion(true);
		System.out.println("mapreduce任务结束……");
		
		System.out.println("");
		System.out.println("查看mapreduce任务结果……");
		System.out.println("单词        数量");
		Scan scan2 = new Scan();
		scan2.setLimit(10);
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.print(Bytes.toString(result.getRow())+"  ");
			System.out.print(Bytes.toString(result.getValue(Bytes.toBytes(HBaseMr.colf), Bytes.toBytes(HBaseMr.col))));
			System.out.println("");
		}
		
		table.close();
		
		System.exit(complet ? 0 : 1);
	}

	

	

	/**
	 * MyMapper 继承 TableMapper TableMapper<Text,IntWritable> Text:输出的key类型，
	 * IntWritable：输出的value类型
	 */
	public static class MyMapper extends TableMapper<Text, IntWritable> {
		private static IntWritable one = new IntWritable(1);
		private static Text word = new Text();

		@Override
		// 输入的类型为：key：rowKey； value：一行数据的结果集Result
		protected void map(ImmutableBytesWritable key, Result value, Context context)
				throws IOException, InterruptedException {
			System.out.println("map " + key);
			// 获取一行数据中的colf：col
			String words = Bytes.toString(value.getValue(Bytes.toBytes(colf), Bytes.toBytes(col)));// 表里面只有一个列族，所以我就直接获取每一行的值
			// 按空格分割
			String itr[] = words.toString().split(" ");
			// 循环输出word和1
			for (int i = 0; i < itr.length; i++) {
				word.set(itr[i]);
				context.write(word, one);
			}
		}
	}

	/**
	 * MyReducer 继承 TableReducer TableReducer<Text,IntWritable> Text:输入的key类型，
	 * IntWritable：输入的value类型， ImmutableBytesWritable：输出类型，表示rowkey的类型
	 */
	public static class MyReducer extends TableReducer<Text, IntWritable, ImmutableBytesWritable> {
		@Override
		protected void reduce(Text key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {
			System.out.println("reduce " + key);
			// 对mapper的数据求和
			int sum = 0;
			for (IntWritable val : values) {// 叠加
				sum += val.get();
			}
			// 创建put，设置rowkey为单词
			Put put = new Put(Bytes.toBytes(key.toString()));
			// 封装数据
			put.addColumn(Bytes.toBytes(colf), Bytes.toBytes(col), Bytes.toBytes(String.valueOf(sum)));
			// 写到hbase,需要指定rowkey、put
			context.write(new ImmutableBytesWritable(Bytes.toBytes(key.toString())), put);
		}
	}

}